package cn.cjxy.dao;

import cn.cjxy.domain.TitleHot;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * Hbase数据访问对象，负责与hbase集群进行交互
 * 主要功能
 * 1. 创建Hbase连接
 * 2. 查询指定时间范围内的电影数据
 * 3. 处理查询结果集
 */
@Repository
public class HBaseDao {

    /**
     * 创建Hbase连接,返回连接对象
     * @return
     */
    private Connection getConnection() throws IOException {

        //获取配置对象
        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
        //本机客户端访问路径
        System.setProperty("hadoop.home.dir", "E:\\hadoop");
        //访问身份
        System.setProperty("HADOOP_USER_NAME", "root");
        //配置zookeeper的地址
        config.set("hbase.zookeeper.quorum", "hugedata01:2181,hugedata02:2181,hugedata03:2181");
        return ConnectionFactory.createConnection(config);
    }

    /**
     * 查询指定时间范围内的电影数据
     * @param start
     * @param end
     * @return
     * @throws IOException
     *
     * 实现逻辑
     *      1.创建时间范围扫描器
     *      2.遍历查询结果集
     *      3.解析cell数据并映射到titlehot对象
     */
    public List<TitleHot> query(long start,long end) throws IOException {
        //构建list
        List<TitleHot> list = new ArrayList<>();
        //建立连接
        Connection connection = getConnection();
        //获取表对象
        Table table = connection.getTable(TableName.valueOf("film_hot"));
        //获取扫描器
        Scan scan = new Scan();
        //设置时间范围
        scan.setTimeRange(start, end);
        //查询结果集
        ResultScanner scanner = table.getScanner(scan);
        //todo   遍历结果集，得到对应结果，映射到实体，放到list中
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()) {
            Result next = iterator.next();//行数据
            //得到单元格集合
            List<Cell> cells = next.listCells();
            TitleHot hot = new TitleHot();
            for (Cell cell : cells) {//一个格子代表一条记录的某列的属性
                //设置id  行键就是id
                hot.setId(Bytes.toString(CellUtil.cloneRow(cell)));
                //获取列名和列值
                String col=Bytes.toString(CellUtil.cloneQualifier(cell));
                String value=Bytes.toString(CellUtil.cloneValue(cell));
                //根据名称匹配
                if ("username".equals(col)) {
                    hot.setUsername(value);
                }else if ("agent".equals(col)) {
                    hot.setAgentType(value);
                }else if ("time".equals(col)) {
                    hot.setTime(Long.parseLong(value));
                }else if ("title".equals(col)) {
                    hot.setTitle(value);
                }else if ("fimltype".equals(col)) {
                    hot.setFilmType(value);
                }else if ("cenimaname".equals(col)) {
                    hot.setCenimaName(value);
                }
                //添加到集合
                list.add(hot);

            }
        }
        //关流
        table.close();
        connection.close();
        return list;
    }

}
